package net.wicp.tams.common.binlog.self.sender.impl.thread;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;

import org.slf4j.Logger;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.common.binlog.self.bean.SendBean;
import net.wicp.tams.common.binlog.self.bean.SendObj;

@Slf4j
public abstract class AbsThread extends Thread {
	protected final String perthreadKey = "_sender";
	protected final boolean serialize;
	protected int sucNum;// 成功处理数
	protected final List<SendObj> errList;// 错误处理记录
	protected final ArrayBlockingQueue<SendBean> queue;

	public AbsThread(int querySize, boolean serialize) {
		this.queue = new ArrayBlockingQueue<>(querySize > 0 ? querySize : 256);
		this.errList = new ArrayList<>();
		this.serialize = serialize;
	}

	@Override
	public void run() {
		while (true) {
			SendBean sendBean = null;
			try {
				sendBean = queue.take();
				doSingle(sendBean);
				sucNum++;
				logSuc(log, sendBean);
			} catch (Throwable e) {
				String beaninfo = sendBean == null ? "" : sendBean.toSendObj().toString();
				log.error("错误对象:[{}],错误原因:[{}]", beaninfo, e.getMessage());
				if (sendBean != null) {
					errList.add(sendBean.toSendObj());
				}
			}
		}
	}

	protected abstract void doSingle(SendBean obj);

	/***
	 * 把上传结果后的信息（ons打messageid，kafka打 分区和移位信息），打印到指定的目录
	 * 
	 * @param log
	 */
	protected abstract void logSuc(Logger log, SendBean obj);

	public void initData() {
		sucNum = 0;
		errList.clear();
	}

	public void addData(SendBean sendBean) throws InterruptedException {
		this.queue.put(sendBean);
	}

}
